//
// Created by tcj on 2022/3/2.
//

#include "iomanager.h"
#include "log.h"
#include "macro.h"


#include "sys/epoll.h"
#include "fcntl.h"
#include "unistd.h"
#include "errno.h"
#include "string"

namespace sylar{
    static sylar::Logger::ptr g_logger =SYLAR_LOG_NAME("system");


    IOManager::FdContext::EventContext& IOManager::FdContext::getContext(Event event){
        switch(event){
            case IOManager::READ:
                return read;
            case IOManager::WRITE:
                return write;
            default:
                SYLAR_ASSERT2(false,"getcontext");
        }
        throw std::invalid_argument("getContext invalid event");
    }

    void IOManager::FdContext::resetContext(EventContext& ctx){
        ctx.scheduler = nullptr;
        ctx.fiber.reset();
        ctx.cb = nullptr;
    }
    void IOManager::FdContext::triggerEvent(Event event){
        SYLAR_ASSERT(events & event);
        events = (Event)(events & ~event);
        EventContext& ctx = getContext(event);
        if(ctx.cb){
            ctx.scheduler->schedule(&ctx.cb);
        }else{
            ctx.scheduler->schedule(&ctx.fiber);
        }

        ctx.scheduler = nullptr;
        return;
    }


    IOManager::IOManager(size_t threads,bool use_caller,
                         const std::string& name)
                         : Scheduler(threads,use_caller,name){
        m_epfd = epoll_create(5000);
        SYLAR_ASSERT(m_epfd>0);

//       这里如果使用信号量，会有阻塞等待信号量，就算有事件触发也不会处理
        int rt = pipe(m_tickleFds);
        SYLAR_ASSERT(!rt);

        epoll_event event;
        memset(&event,0,sizeof(epoll_event));
        // 表示当前事件触发模式是边缘触发
//        水平触发：只要缓冲区中有数据，就一直通知
//        边缘触发：缓冲区中有数据指挥通知一次，之后在有数据才会通知（若是读数据的时候没有读完，则生于的数据不会再通知，知道有新的数据到来）
        event.events = EPOLLIN | EPOLLET;
        // [0]:为读管道   [1]：为写管道   设置读事件触发
        event.data.fd = m_tickleFds[0];

        //非阻塞方式，配合边缘触发
        //fcntl是用来修改已经打开文件属性  设置管道标识符为非阻塞
//  如果是非阻塞，read函数运行时，会先看一下管道中是否有数据，如果有数据，则正常运行读取数据，如果管道中没有数据，则read函数会立即返回
        rt = fcntl(m_tickleFds[0],F_SETFL,O_NONBLOCK);
        SYLAR_ASSERT(!rt);

        rt = epoll_ctl(m_epfd,EPOLL_CTL_ADD,m_tickleFds[0],&event);
        SYLAR_ASSERT(!rt);

        contextResize(32);
//        m_fdContexts.resize(64);
        start();

    }


//   对于析沟函数，首先要等scheduler调度完所有的任务，然后再关闭epoll句柄和pipe句柄
//    然后释放所有的fdcontext

    IOManager::~IOManager(){
        stop();
        close(m_epfd);
        close(m_tickleFds[0]);
        close(m_tickleFds[1]);

        for(size_t i=0;i<m_fdContexts.size();++i){
            if(m_fdContexts[i]){
                delete m_fdContexts[i];
            }
        }
    }

    // 修改句柄数量
    void IOManager::contextResize(size_t size) {
        m_fdContexts.resize(size);
        for(size_t i=0;i<m_fdContexts.size();++i){
            if(!m_fdContexts[i]){
                m_fdContexts[i] = new FdContext;
                m_fdContexts[i]->fd =i;
            }
        }
    }

//    fd描述符发生event事件执行回调函数
//
    //1.success  0.retry  -1.error
    int IOManager::addEvent(int fd,Event event,std::function<void()> cb){
        FdContext* fd_ctx = nullptr;
        // 拿到对应的句柄对象  没有就创建
        RWMutexType::ReadLock lock(m_mutex);
        if((int)m_fdContexts.size() > fd){
            fd_ctx = m_fdContexts[fd];
            lock.unlock();
        }else{
            lock.unlock();
            RWMutexType::WriteLock lock2(m_mutex);
            contextResize(fd*1.5);
            fd_ctx = m_fdContexts[fd];
        }

        //同一个fd不允许重复添加相同的事件
        FdContext::MutexType::Lock lock2(fd_ctx->mutex);
        if(fd_ctx->events & event){
            SYLAR_LOG_ERROR(g_logger)<<"addevent assert fd="<<fd
                        <<" event="<<event
                        <<" fd_ctx.events="<<fd_ctx->events<<std::endl;
            SYLAR_ASSERT(!(fd_ctx->events & event));
        }

        int op = fd_ctx->events?EPOLL_CTL_MOD :EPOLL_CTL_ADD;
        epoll_event epevent;
        epevent.events = EPOLLET | fd_ctx->events | event;
        epevent.data.ptr = fd_ctx;

        int rt = epoll_ctl(m_epfd,op,fd,&epevent);

        if(rt){
            SYLAR_LOG_ERROR(g_logger)<<"epoll_ctl("<<m_epfd
                    <<op<<","<<fd<<","<<epevent.events<<");"
                    <<rt<<"("<<errno<<") ("<< strerror(errno)<<")";
            return -1;
        }

        //待处理事件子赠
        ++m_pendingEventCount;
        // 将句柄上的事件进行叠加
        fd_ctx->events = (Event)(fd_ctx->events | event);
        //
        FdContext::EventContext& event_ctx = fd_ctx->getContext(event);
        SYLAR_ASSERT( (!event_ctx.scheduler)
                    &&(!event_ctx.fiber)
                    && (!event_ctx.cb));
//        赋值scheduler和回调函数，如果回调函数为空，则将当前协程当成回调执行体
        event_ctx.scheduler = Scheduler::GetThis();

        if(cb){
            event_ctx.cb.swap(cb);
        }else{
            event_ctx.fiber = Fiber::GetThis();
            SYLAR_ASSERT(event_ctx.fiber->getState() == Fiber::EXEC);
        }
        return 0;
    }

    bool IOManager::delEvent(int fd,Event event){
        RWMutexType::ReadLock lock(m_mutex);
        if((int)m_fdContexts.size()<=fd){
            return false;
        }
        FdContext* fd_ctx = m_fdContexts[fd];
        lock.unlock();

        FdContext::MutexType::Lock lock2(fd_ctx->mutex);
        if(!(fd_ctx->events & event)){
            return false;
        }
    //  清除指定的事件，表示不关心这个事件，如果清除之后结果为0，则从epoll_wait删除该文件描述符
        Event new_events = (Event)(fd_ctx->events & ~event);
        int op = new_events?EPOLL_CTL_MOD : EPOLL_CTL_DEL;
        epoll_event epevent;
        epevent.events =EPOLLET | new_events;
        epevent.data.ptr = fd_ctx;

        int rt = epoll_ctl(m_epfd,op,fd,&epevent);
        if(rt){
            SYLAR_LOG_ERROR(g_logger)<<"epoll_ctl("<<m_epfd
             <<op<<","<<fd<<","<<epevent.events<<");"
             <<rt<<"("<<errno<<") ("<< strerror(errno)<<")";
            return false;
        }
        --m_pendingEventCount;
        //重置该fd对应的event事件上下文
        fd_ctx->events = new_events;
        FdContext::EventContext& event_ctx = fd_ctx->getContext((event));
        fd_ctx->resetContext(event_ctx);
        return true;
    }

//    如果该事件被注册过回调，那就触发一次回调函数
    bool IOManager::cancelEvent(int fd,Event event){
        RWMutexType::ReadLock lock(m_mutex);
        if((int)m_fdContexts.size()<=fd){
            return false;
        }
        FdContext* fd_ctx = m_fdContexts[fd];
        lock.unlock();

        FdContext::MutexType::Lock lock2(fd_ctx->mutex);
        if(!(fd_ctx->events & event)){
            return false;
        }
        Event new_events = (Event)(fd_ctx->events & ~event);
        int op = new_events?EPOLL_CTL_MOD : EPOLL_CTL_DEL;
        epoll_event epevent;
        epevent.events =EPOLLET | new_events;
        epevent.data.ptr = fd_ctx;

        int rt = epoll_ctl(m_epfd,op,fd,&epevent);
        if(rt){
            SYLAR_LOG_ERROR(g_logger)<<"epoll_ctl("<<m_epfd
                                     <<op<<","<<fd<<","<<epevent.events<<");"
                                     <<rt<<"("<<errno<<") ("<< strerror(errno)<<")";
            return false;
        }

        fd_ctx->triggerEvent(event);
        --m_pendingEventCount;
        return true;
    }

    bool IOManager::cancalAll(int fd){
        RWMutexType::ReadLock lock(m_mutex);
        if((int)m_fdContexts.size()<=fd){
            return false;
        }
        FdContext* fd_ctx = m_fdContexts[fd];
        lock.unlock();

        FdContext::MutexType::Lock lock2(fd_ctx->mutex);
        if(!fd_ctx->events){
            return false;
        }
        int op = EPOLL_CTL_DEL;
        epoll_event epevent;
        epevent.events =0;
        epevent.data.ptr = fd_ctx;

        int rt = epoll_ctl(m_epfd,op,fd,&epevent);
        if(rt){
            SYLAR_LOG_ERROR(g_logger)<<"epoll_ctl("<<m_epfd
                                     <<op<<","<<fd<<","<<epevent.events<<");"
                                     <<rt<<"("<<errno<<") ("<< strerror(errno)<<")";
            return false;
        }

//        触发全部已经注册的事件
        if(fd_ctx->events & READ){
            fd_ctx->triggerEvent(READ);
            --m_pendingEventCount;
        }

        if(fd_ctx->events & WRITE){
            fd_ctx->triggerEvent(WRITE);
            --m_pendingEventCount;
        }

        SYLAR_ASSERT(fd_ctx->events ==0);
        return true;
    }
    IOManager* IOManager::GetThis(){
        return dynamic_cast<IOManager*>(Scheduler::GetThis());
    }

    //通知任务调度器有任务要调度
    void IOManager::tickle(){
        if(hasIdleThreads()){
            return;
        }
        //写pipe让idle协程从epoll_wait退出
        //待idle协程yield之后，schedule::run就可以调度其他任务了
        int rt = write(m_tickleFds[1],"T",1);
        SYLAR_ASSERT(rt==1);
    }

    bool IOManager::stopping(){
        uint64_t  timeout=0;
        return stopping(timeout);
    }
// 调度器停止的判断条件
    bool IOManager::stopping(uint64_t &timeout) {
        timeout = getNextTimer();
        return timeout==0
            &&m_pendingEventCount==0
            && Scheduler::stopping();
    }


//io调度协程，阻塞应该在等待io事件上，idle退出的时机就是epoll_wait返回
//  对应的操作应该时tickle或者注册的io事件就绪
/**
 * @brief idle协程
 * @details 对于IO协程调度来说，应阻塞在等待IO事件上，idle退出的时机是epoll_wait返回，
 对应的操作是tickle或注册的IO事件就绪
 * 调度器无调度任务时会阻塞idle协程上，对IO调度器而言，idle状态应该关注两件事，
 一是有没有新的调度任务，对应Schduler::schedule()，如果有新的调度任务，
             那应该立即退出idle状态，并执行对应的任务；
 二是关注当前注册的所有IO事件有没有触发，如果有触发，那么应该执行IO事件对应的回调函数
 */
    void IOManager::idle(){

        //一次epoll_wait最多检测64个就绪事件
        epoll_event* events = new epoll_event[64]();
        std::shared_ptr<epoll_event> shared_events(events, [](epoll_event* ptr){
            delete[] ptr;
        });
        //调度器没有任务时，会阻塞在idle协程上
        while(true){
            //获取下一个定时器的时间，
            //顺便判断调度器是否结束
            uint64_t next_timeout=0;
            if(stopping(next_timeout)){
//                next_timeout = getNextTimer();
//                if(next_timeout==~0ull){
                SYLAR_LOG_INFO(g_logger)<<"name="
                                <<getName()
                                <<" idle stopping exit";
                break;
//                }
            }

            int rt=0;

            do{
                // 超时时间考虑最近快到期的定时任务和默认超时时间取min选其中最小值
                static const int MAX_TIMEOUT = 5000;
                if(next_timeout !=~0ull){
                    next_timeout = (int)next_timeout>MAX_TIMEOUT
                                    ?MAX_TIMEOUT:next_timeout;
                }else{
                    next_timeout = MAX_TIMEOUT;
                }
                //直接阻塞epoll_wait上，等待事件的发生
                rt = epoll_wait(m_epfd,events,64,(int)next_timeout);
//                SYLAR_LOG_INFO(g_logger)<<"epoll_wait"<<rt;
                if(rt<0 && errno==EINTR){
                    // errno=EINTR：表示文件在读和写的过程中出现了中断错误
                }else{
                    break;
                }
            }while(true);

            // 检查是否有超时定时器需要处理，需要的话就将超时的任务加入调度任务中
            std::vector<std::function<void()>> cbs;
            listExpiredCb(cbs);
            if(!cbs.empty()){
                SYLAR_LOG_INFO(g_logger)<<"on timer sb.size()"
                    <<cbs.size();
                schedule(cbs.begin(),cbs.end());
                cbs.clear();
            }

            for(int i=0;i<rt;++i){
                epoll_event &event = events[i];
                // 如果有管道读io，仅仅是将其唤醒，忽略
                if(event.data.fd == m_tickleFds[0]){
                    uint8_t dummy[256];
                    while(read(m_tickleFds[0],dummy,sizeof(dummy))>0);
                    continue;
                }
                // 网络io,依次轮询触发注册好的异步读/写回调函数作为任务加入队列供调度
                //通过epoll_event的私有指针获取Fdcontext
                FdContext* fd_ctx = (FdContext *)event.data.ptr;
                FdContext::MutexType::Lock lock(fd_ctx->mutex);

//                EPOLLERR:出错，比如写读端已经关闭pipe
//                EPOLLHUP:套接字对端关闭
//                出现这两种事件，应该同时触发fd的读和写事件， 否则有可能出现注册的事件永远执行不到的情况
                if(event.events & (EPOLLERR | EPOLLHUP)){
                    event.events |= (EPOLLIN | EPOLLOUT) & fd_ctx->events;
                }
                int real_events = NONE;
                if(event.events & EPOLLIN){
                  real_events |= READ;
                }
                if(event.events & EPOLLOUT){
                    real_events |= WRITE;
                }

                if((fd_ctx->events & real_events)==NONE){
                    continue;
                }
                // 上述得到的事件进行重新的加入或者删除epoll树上
                //删除已经发生的事件，将剩余的事件重新加入epoll_wait
                //如果剩余事件为0，表示这个fd已经不需要关注了，直接从epoll中删除
                int left_events = (fd_ctx->events & ~real_events);
                int op = left_events?EPOLL_CTL_MOD:EPOLL_CTL_DEL;
                event.events = EPOLLET |left_events;

                int rt2 = epoll_ctl(m_epfd,op,fd_ctx->fd,&event);
                if(rt2){
                    SYLAR_LOG_ERROR(g_logger)<<"epoll_ctl("<<m_epfd
                             <<op<<","<<fd_ctx->fd<<","<<(EPOLL_EVENTS)event.events<<");"
                             <<rt2<<"("<<errno<<") ("<< strerror(errno)<<")";
                    continue;
                }
//                //处理已经发生的事件，就是让调度器调度指定的函数或者协程
//                triggerevent:实际上只是把对应的fiber重新加入调度，要执行的话，要等idle协程退出
                if(real_events & READ){
                    fd_ctx->triggerEvent(READ);
                    --m_pendingEventCount;
                }

                if(real_events & WRITE){
                    fd_ctx->triggerEvent(WRITE);
                    --m_pendingEventCount;
                }
            }
//            一旦处理完所有事件，idle协程yield，这样就可以让协程调度（run）检查是否有新任务要调度
            Fiber::ptr cur = Fiber::GetThis();
            auto raw_ptr = cur.get();
            cur.reset();

            raw_ptr->swapOut();
        }
    }

    // 定时器队列头插入对象后进行epoll_wait超时更新
    void IOManager::onTimerInsertedAtFront() {

        tickle();

    }
}